package mspbots.hubspot.client.core.client;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.log4j.Log4j2;
import mspbots.hubspot.client.annotation.ApiException;
import mspbots.hubspot.client.oauth.AuthService;
import mspbots.hubspot.common.QueryBody;
import mspbots.hubspot.common.RabbitKeys;
import mspbots.hubspot.common.ResultSync;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.util.MultiValueMap;
import org.springframework.util.ObjectUtils;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;

/**
 * data.connector.hubspot.client.HubSpotClient
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2020/4/3
 */
@Log4j2
@Service
public class HubSpotClient {

    private final WebClient webClient;
    private final RabbitMessagingTemplate rabbitMessagingTemplate;
    private final ObjectMapper objectMapper;
    private final AuthService authService;

    public HubSpotClient(WebClient.Builder builder,
                         RabbitMessagingTemplate rabbitMessagingTemplate,
                         ObjectMapper objectMapper, AuthService authService) {
        this.webClient = builder.baseUrl("https://api.hubapi.com/")
                .defaultHeader("Accept", "application/json").build();
        this.rabbitMessagingTemplate = rabbitMessagingTemplate;
        this.objectMapper = objectMapper;
        this.authService = authService;
    }


    public Mono<JsonNode> get(String path, MultiValueMap<String, String> queryParams) {
        return Mono.subscriberContext().flatMap(ctx -> this.authService.getToken(ctx.get("tenantId"))
                .flatMap(authorization -> this.webClient.get()
                        .uri(builder -> builder.path(path).queryParams(queryParams).build())
                        .headers(httpHeaders -> httpHeaders.setBearerAuth(authorization.getAccessToken()))
                        .retrieve()
                        .onStatus(HttpStatus::isError, clientResponse -> clientResponse.bodyToMono(JsonNode.class)
                                .map(ApiException::withMsg))
                        .bodyToMono(JsonNode.class))
        );
    }

    public Mono<JsonNode> put(String path, MultiValueMap<String, String> queryParams) {
        return Mono.subscriberContext().flatMap(ctx -> this.authService.getToken(ctx.get("tenantId"))
                .flatMap(authorization -> this.webClient.put()
                        .uri(builder -> builder.path(path).queryParams(queryParams).build())
                        .headers(httpHeaders -> httpHeaders.setBearerAuth(authorization.getAccessToken()))
                        .retrieve()
                        .onStatus(HttpStatus::isError, clientResponse -> clientResponse.bodyToMono(JsonNode.class)
                                .map(ApiException::withMsg))
                        .bodyToMono(JsonNode.class))
        );
    }

    public Mono<Object> delete(String path) {
        return Mono.subscriberContext().flatMap(ctx -> this.authService.getToken(ctx.get("tenantId"))
                .flatMap(authorization -> this.webClient.delete()
                        .uri(builder -> builder.path(path).build())
                        .headers(httpHeaders -> httpHeaders.setBearerAuth(authorization.getAccessToken()))
                        .retrieve()
                        .onStatus(HttpStatus::isError, clientResponse -> clientResponse.bodyToMono(JsonNode.class)
                                .map(ApiException::withMsg))
                        .bodyToMono(Void.class)));
    }

    public Mono<JsonNode> patch(String path, Object body) {
        try {
            log.debug("Patch [{}],Body: {}", path, this.objectMapper.writeValueAsString(body));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return Mono.subscriberContext().flatMap(ctx -> this.authService.getToken(ctx.get("tenantId"))
                .flatMap(authorization -> this.webClient.patch()
                        .uri(builder -> builder.path(path).build())
                        .headers(httpHeaders -> httpHeaders.setBearerAuth(authorization.getAccessToken()))
                        .bodyValue(body)
                        .retrieve()
                        .onStatus(HttpStatus::isError, clientResponse -> clientResponse.bodyToMono(JsonNode.class)
                                .map(ApiException::withMsg))
                        .bodyToMono(JsonNode.class)));
    }

    public Mono<JsonNode> post(String path, Object body) {
        try {
            log.debug("Post [{}],Body: {}", path, this.objectMapper.writeValueAsString(body));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return Mono.subscriberContext().flatMap(ctx -> this.authService.getToken(ctx.get("tenantId"))
                .flatMap(authorization -> {
                    ResultSync resultSync = ctx.getOrDefault("resultSync", ResultSync.withDefault(path));
                    resultSync.setElapsedTime(System.currentTimeMillis());
                    resultSync.setTenantId(authorization.getTenantId());
                    resultSync.setParams(this.objectMapper.convertValue(body, JsonNode.class));
                    return this.webClient.post().uri(builder -> builder.path(path).build())
                            .headers(httpHeaders -> httpHeaders.setBearerAuth(authorization.getAccessToken()))
                            .bodyValue(body)
                            .retrieve()
                            .onStatus(HttpStatus::isError, clientResponse -> clientResponse.bodyToMono(JsonNode.class)
                                    .map(ApiException::withMsg))
                            .bodyToMono(JsonNode.class)
                            .doOnNext(resp -> this.handlerSuccessNext(resp, resultSync))
                            .onErrorResume(err -> {
                                this.handlerException(err, resultSync);
                                return Mono.error(err);
                            });
                })
        );
    }

    public Flux<JsonNode> fetch(String path, QueryBody queryBody) {
        return this.post(path, queryBody)
                .delayElement(Duration.ofMillis(100))
                .expand(jsonNode -> this.autoNextPage(jsonNode, path, queryBody))
                .flatMap(jsonNode -> Flux.fromIterable(jsonNode.withArray("results")));
    }

    private Mono<JsonNode> autoNextPage(JsonNode resp, String path, QueryBody queryBody) {

        JsonNode pageDetailsNode = resp.get("paging");

        if (ObjectUtils.isEmpty(pageDetailsNode) || resp.get("total").intValue() == 0) {
            return Mono.empty();
        }

        String afterStr = pageDetailsNode.get("next").get("after").textValue();
        int after = Integer.parseInt(afterStr);
        if (after == 0) {
            return Mono.empty();
        }
        queryBody.setAfter(after);
        return this.post(path, queryBody);

    }

    private void handlerException(Throwable err, ResultSync resultSync) {
        resultSync.setElapsedTime(System.currentTimeMillis() - resultSync.getElapsedTime());
        if (err instanceof ApiException) {
            resultSync.setMsg(((ApiException) err).getMsg());
        } else {
            resultSync.setMsg(err.getMessage());
        }
        resultSync.setStatus(500);

        log.error("Request {} error action {}, elapsedTime {}, msg: {}",
                resultSync.getTenantId(), resultSync.getAction(),
                resultSync.getElapsedTime(), resultSync.getMsg());

        this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                RabbitKeys.LOGS_TASK_QUEUE_KEY, resultSync);
    }

    private void handlerSuccessNext(JsonNode resp, ResultSync resultSync) {
        JsonNode pageDetailsNode = resp.get("total");
        if (!ObjectUtils.isEmpty(pageDetailsNode)) {
            resultSync.setCount(pageDetailsNode.intValue());
        } else {
            resultSync.setCount(resp.size());
        }
        resultSync.setElapsedTime(System.currentTimeMillis() - resultSync.getElapsedTime());

        log.info("Request {} success, action {}, elapsedTime {}, count {}",
                resultSync.getTenantId(), resultSync.getAction(),
                resultSync.getElapsedTime(), resultSync.getCount());

        this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                RabbitKeys.LOGS_TASK_QUEUE_KEY, resultSync);
    }
}
